package com.doit.day02;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Properties;

public class ConsumerDemo3 {
    public static void main(String[] args) {
        Properties props = new Properties();
        /**
         * 必配的
         */
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux01:9092,linux02:9092,linux03:9092");
        //key和value的反序列化方式
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //消费者组
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "idea_group01");

        /**
         * 选配的
         */
        //是否允许自动创建topic
        props.setProperty("allow.auto.create.topics", "true");
        //latest, earliest, none
        // 如果说，有组id，但是你已经指定了从什么地方开始消费，那么他就会从你指定的位置开始消费
        //如果说你没有指定位置开始消费，会去__consumer_offset 这个topic里面有之前的记录，那么就会失效，从上次消费到的地方开始消费
        // 如果没有，那么会按照下面设置的这个参数来执行
        props.setProperty("auto.offset.reset", "earliest");

        //是否允许自动提交偏移量  __consumer_offset  全自动
        props.setProperty("enable.auto.commit", "false");  //如果是true，有一定的可能会丢数据
        //如果设置成了false 那么他就不自动提交偏移量了  ==》手动提交   半自动  commitAsync  commitSync ==》 同样的是提交到  __consumer_offset
        props.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.CooperativeStickyAssignor");

        //能否实现精准一次性消费呢？？？？


        //自动提交偏移量，每隔多长时间提交一次
//        props.setProperty("auto.commit.interval.ms","10000");

        //创建kafka的消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        //订阅主题  第一种方式
        consumer.subscribe(Arrays.asList("reb-1","reb-2"), new ConsumerRebalanceListener() {
            /**
             * 回收前我每个消费者消费的是哪些分区的方案
             * @param partitions
             */
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                System.out.println("我是原来的方案：" + partitions);

            }

            /**
             * 发生了在均衡以后，我是一个什么样的方案
             * @param partitions
             */
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                System.out.println("我是在均衡之后的方案：" + partitions);
                System.out.println("===============分界线=====================");

            }
        });


        while (true) {
            //拉取broker中的数据
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));

            //处理数据的业务逻辑
            for (ConsumerRecord<String, String> consumerRecord : poll) {
//                String value = consumerRecord.value();
//                String topic = consumerRecord.topic();
//                int partition = consumerRecord.partition();
//                long offset = consumerRecord.offset();
//                /**
//                 * 时间戳类型有两种：
//                 *  1.数据写入到kafka中的时间（创建这条数据的时间）  默认
//                 *  2.数据追加到日志里面的时间戳
//                 */
//                long timestamp = consumerRecord.timestamp();
//                TimestampType timestampType = consumerRecord.timestampType();
//                System.out.println("主题：" + topic + ",分区：" + partition + ",偏移量:" + offset + ",值：" + value + ",时间戳：" + timestamp + ",时间戳类型:" + timestampType);
            }

            //同步提交偏移量
            consumer.commitSync();

        }


    }
}
